﻿using Microsoft.Extensions.DependencyInjection;
using Stee.CAP8.MessageBus.Interface.Enums;
using Stee.CAP8.MessageBus.Interface.Interfaces;
using Stee.CAP8.MessageBus.Interface.Models;
using Stee.CAP8.MessageBus.Kafka.Configurations;
using Stee.CAP8.MessageBus.Kafka.Publisher;
using Stee.CAP8.MessageBus.Kafka.Subscriber;
using System;
using System.Collections.Generic;
using System.Threading;

namespace test.kafka
{
    public class EventBus : IBus 
    {
        #region private fileds
        private static readonly string exitCommand = "EXIT";

        private static readonly string topic = "first_topic";
        private static readonly string key = "first_key";
        private static readonly string groupID = "first_group";

        private static readonly string topic2 = "second_topic";
        private static readonly string key2 = "second_key";
        private static readonly string groupID2 = "second_group";

        private static ISubscriber<string> subscriber;

        private static readonly TopicName topicNameString = new TopicName(topic, key, MessageType.Event);
        private static readonly TopicName topicNameObject = new TopicName(topic2, key2, MessageType.Event);


        private static bool isExit = false;

        SubscriberConfiguration subscriberConfig = new SubscriberConfiguration(new Dictionary<string, string>
                {
                    { "BootstrapServers", "127.0.0.1:9092" },
                    { "GroupID", "first_group" },
                    { "ClientID", "mysubscriber" },
                    { "EnableAutoCommit", "true" },
                    { "AutoCommitIntervalMs", "5000" },
                    { "EnableAutoOffsetStore", "true" },
                });

        PublisherConfiguration publishConfig = new PublisherConfiguration(new Dictionary<string, string>
                {
                    { "BootstrapServers", "127.0.0.1:9092" },
                    { "Acks", "All" },
                    { "ClientID", "mypublisher" },
                    { "EnableIdempotence", "true" },
                    { "LingerMs", "0.5" },
                    { "MessageSendMaxRetries", "2" },
                    { "MessageTimeoutMs", "300000" },
                    { "QueueBufferingMaxKbytes", "1048576" },
                    { "QueueBufferingMaxMessages", "100000" },
                });

        #endregion

        public EventBus() 
        {

        }


        public void Subscribe<TEventMessage>( IEventHandler<TEventMessage> handler)
            where TEventMessage: IEventMessage

        {
            var subscriber = new SubscriberImpl<TEventMessage>(topicNameString, subscriberConfig,(topicName,msgObject)=> {
                Console.WriteLine($"TopicName is {topicName}");
                handler.Handle(msgObject);
                return true;
            });
            subscriber.Subscribe();
        }


        public void Publish<TEventMessage>(TEventMessage message)
           where TEventMessage : IEventMessage

        {
            var publisher = new PublisherImpl<TEventMessage>(topicNameString, publishConfig, (msgObject) => {
                Console.WriteLine($"发送的消息是{msgObject}");

                return true;
            });
            publisher.Publish(message,CancellationToken.None);
        }


    }

   
}
